package com.atguigu.Func;

import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.util.DimUtil_lj;
import com.atguigu.util.DruidDSUtil_lj;
import com.atguigu.util.ThreadPoolUtil_lj;
import lombok.SneakyThrows;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @Author 晶晶
 * @Date 2023 03 27 16 13
 **/
public abstract class AsyncDimFunction_lj <T> extends RichAsyncFunction<T, T> implements AsyncJoinFunction_lj<T> {
    private ThreadPoolExecutor threadPoolExecutor;
    private String tableName;

    public AsyncDimFunction_lj(String tableName) {
        this.tableName = tableName;
    }

    //初始化线程池
    @Override
    public void open(Configuration parameters) throws Exception {
        threadPoolExecutor= ThreadPoolUtil_lj.getThreadPoolExecutor();
    }

    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) throws Exception {
        threadPoolExecutor.execute(new Runnable() {
            @SneakyThrows
            @Override
            public void run() {

                //获取维度信息
                //通过连接池获取链接
                DruidPooledConnection conn = DruidDSUtil_lj.getPhoenixConn();
                String key =getKey(input);
                JSONObject dimInfo = DimUtil_lj.getDimInfo(conn, tableName,key);

                //补充字段
                if (dimInfo!=null) {
                    join(input, dimInfo);
                }
                conn.close();

                //输出
                resultFuture.complete(Collections.singleton(input));
            }

        });
    }

    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        System.out.println("timeout data:" + input);
    }
}
